/*
 * TcpConnection.cpp
 *
 *  Created on: May 9, 2011
 *      Author: michael
 */

#include "../StdAfx.h"
#include "TcpConnection.h"
#include "RawData.h"
#include <boost/bind.hpp>

using namespace boost::asio;

namespace DerbyServ
{

TcpConnection::TcpConnection(
		io_service & ioService_
		, int connectionId_)
	: _socket(ioService_)
	, _id(connectionId_)
	, _writingStarted(false)
	, _readingStarted(false)
	, _writingMode(MODE_META)
	, _readingMode(MODE_META)
{
}


TcpConnection::~TcpConnection()
{
}


TcpConnectionPtr TcpConnection::Create(io_service & ioService_, int connectionId)
{
	TcpConnectionPtr ptr( new TcpConnection(ioService_, connectionId)); //ioService cannot be const
	return ptr;
}

ip::tcp::socket & TcpConnection::GetSocket()
{
	return _socket;
}

int TcpConnection::GetId() const
{
	return _id;
}

/**
 * !! This method blocks if _writingEvents is empty !!
 */
void TcpConnection::WriteAsync()
{
	LOG_SCOPE;
	boost::mutex::scoped_lock scoped_lock(_writingMutex);

	if (_writingMode == MODE_META && !_writingRaws.empty())
	{
		_writingRaw = _writingRaws.front();
		_writingRaws.pop();

		BOOST_ASSERT(_writingRaw.Size() == (int)_writingRaw.Data().size());

		_writingRaw.SerializeMeta();
		RawType & writingMeta = _writingRaw.SerializedMeta();
		BOOST_ASSERT(writingMeta.size() > 0);

		async_write(
				_socket,
				buffer(writingMeta),
				boost::bind(&TcpConnection::HandleWrite, shared_from_this(),
						placeholders::error, placeholders::bytes_transferred));
		_writingStarted = true;

	}
	else if (_writingMode == MODE_DATA)
	{
		RawType & writingData = _writingRaw.Data();
		BOOST_ASSERT(writingData.size() > 0);
		BOOST_ASSERT(_writingRaw.Size() == (int)_writingRaw.Data().size());

		async_write(
				_socket,
				buffer(writingData),
				boost::bind(&TcpConnection::HandleWrite, shared_from_this(),
						placeholders::error, placeholders::bytes_transferred));
		_writingStarted = true;
	}
}

void TcpConnection::HandleWrite(
		const boost::system::error_code & error, std::size_t bytes_transferred)
{
	LOG_SCOPE;

	{
		boost::mutex::scoped_lock scoped_lock(_writingMutex);
		BOOST_ASSERT(_writingStarted);
		_writingStarted = false;

		if (!error)
		{
			_writingMode = _writingMode == MODE_META ? MODE_DATA : MODE_META;
		}
	}

	if (!error)
	{
		WriteAsync();
	}
	else
	{
		LOGE("Writing to connection #" << GetId() << " failed. ");
				//TODO disconnect
	}
}


void TcpConnection::ReadAsync()
{
	LOG_SCOPE;
	boost::mutex::scoped_lock scoped_lock(_readingMutex);

	if (_readingMode == MODE_META)
	{
		RawType & readingMeta = _readingRaw.SerializedMeta();
		BOOST_ASSERT(readingMeta.size() == RawData::META_BYTE_SIZE);

		async_read(
				_socket,
				buffer(readingMeta),
				boost::bind(&TcpConnection::HandleRead, shared_from_this(),
						placeholders::error));
		_readingStarted = true;
	}
	else if (_readingMode == MODE_DATA)
	{
		BOOST_ASSERT(_readingRaw.Size() != 0);
		BOOST_ASSERT(_readingRaw.Size() < 1024);
		RawType & readingData = _readingRaw.Data();
		readingData.resize(_readingRaw.Size());

		async_read(
				_socket,
				buffer(readingData),
				boost::bind(&TcpConnection::HandleRead, shared_from_this(),
						placeholders::error));
		_readingStarted = true;
	}
}

void TcpConnection::HandleRead(
		const boost::system::error_code & error)
{
	LOG_SCOPE;

	{
		boost::mutex::scoped_lock scoped_lock(_readingMutex);
		_readingStarted = false;

		if (!error)
		{
			if (_readingMode == MODE_META)
			{
				_readingRaw.DeserializeMeta();
				_readingMode = MODE_DATA;
			}
			else if (_readingMode == MODE_DATA)
			{
				BOOST_ASSERT(_readingRaw.Size() == (int)_readingRaw.Data().size());
				_readingRaws.push(_readingRaw);
				_readingMode = MODE_META;
			}
		}
	}

	if (!error)
	{
		ReadAsync();
	}
	else
	{
		LOGE("Reading from connection #" << GetId() << "failed. ");
		//TODO disconnect
	}
}


bool TcpConnection::Send(const RawData & raw)
{
	LOG_SCOPE;
	{
		boost::mutex::scoped_lock scoped_lock(_writingMutex);
		_writingRaws.push(raw);
	}

	if (!_writingStarted) //TODO ?? thread-safe ??
	{
		WriteAsync();
	}
	return true;
}

bool TcpConnection::Receive(RawData & raw)
{
	bool retval = false;

	{
		boost::mutex::scoped_lock scoped_lock(_readingMutex);

		if (!_readingRaws.empty())
		{
			LOGD("TcpConnection::Receive (!_readingEvents.empty()) is true");
			raw = _readingRaws.front();
			_readingRaws.pop();
			retval = true;
		}
	}

	if (!_readingStarted) //TODO ?? thread-safe ??
	{
		ReadAsync();
	}
	return retval;
}


} //namespace
